using System;
using System.Collections.Generic;
using ByteBuffer = Me.Prettyprint.Cassandra.Utils.ByteBuffer;
using Me.Prettyprint.Hector.Api;
using Org.Apache.Cassandra.Thrift;

namespace Me.Prettyprint.Cassandra.Service
{
	public sealed class BatchMutation<K>
	{
		public BatchMutation(Serializer<K> serializer, BatchSizeHint sizeHint)
		{
            KeySerializer = serializer;
            SizeHint = sizeHint;
            if (null == sizeHint)
            {
                MutationDictionary = new Dictionary<ByteBuffer, Dictionary<string, List<Mutation>>>();
            }
            else
            {
                MutationDictionary = new Dictionary<ByteBuffer, Dictionary<string, List<Mutation>>>(sizeHint.NumOfRows);
            }
		}
        public BatchMutation(Serializer<K> serializer)
            : this(serializer, null)
        {
        }
        private BatchMutation(Serializer<K> serializer, Dictionary<ByteBuffer, Dictionary<string, List<Mutation>>> mutationMap, BatchSizeHint sizeHint)
        {
            KeySerializer = serializer;
            MutationDictionary = mutationMap;
            SizeHint = sizeHint;
        }

        public Dictionary<ByteBuffer, Dictionary<string, List<Mutation>>> MutationDictionary
        {
            get;
            protected set;
        }
        public Serializer<K> KeySerializer
        {
            get;
            private set;
        }
        public BatchSizeHint SizeHint
        {
            get;
            private set;
        }
        public BatchMutation<K> AddInsertion(K key, string columnFamily, Column column)
        {
            Mutation mutation = new Mutation();
            mutation.Column_or_supercolumn = new ColumnOrSuperColumn();
            mutation.Column_or_supercolumn.Column = column;
            List<string> columnFamilies = new List<string>(1);
            columnFamilies.Add(columnFamily);
            AddMutation(key, columnFamilies, mutation);
            return this;
        }
        public BatchMutation<K> AddInsertion(K key, List<string> columnFamilies, Column column)
        {
            Mutation mutation = new Mutation();
            mutation.Column_or_supercolumn = new ColumnOrSuperColumn();
            mutation.Column_or_supercolumn.Column = column;
            AddMutation(key, columnFamilies, mutation);
            return this;
        }
        public BatchMutation<K> AddSuperInsertion(K key, List<string> columnFamilies, SuperColumn superColumn)
        {
            Mutation mutation = new Mutation();
            mutation.Column_or_supercolumn = new ColumnOrSuperColumn();
            mutation.Column_or_supercolumn.Super_column = superColumn;
            AddMutation(key, columnFamilies, mutation);
            return this;
        }
        public BatchMutation<K> AddCounterInsertion(K key, List<string> columnFamilies, CounterColumn counterColumn)
        {
            Mutation mutation = new Mutation();
            mutation.Column_or_supercolumn = new ColumnOrSuperColumn();
            mutation.Column_or_supercolumn.Counter_column = counterColumn;
            AddMutation(key, columnFamilies, mutation);
            return this;
        }
        public BatchMutation<K> AddSuperCounterInsertion(K key, List<string> columnFamilies, CounterSuperColumn counterSuperColumn)
        {
            Mutation mutation = new Mutation();
            mutation.Column_or_supercolumn = new ColumnOrSuperColumn();
            mutation.Column_or_supercolumn.Counter_super_column = counterSuperColumn;
            AddMutation(key, columnFamilies, mutation);
            return this;
        }
        public BatchMutation<K> AddDeletion(K key, List<string> columnFamilies, Deletion deletion)
        {
            Mutation mutation = new Mutation();
            mutation.Deletion = deletion;
            AddMutation(key, columnFamilies, mutation);
            return this;
        }
        private void AddMutation(K key, List<string> columnFamilies, Mutation mutation)
        {
            Dictionary<string, List<Mutation>> innerMutationMap = GetInnerMutationMap(key);
            foreach (string columnFamily in columnFamilies)
            {
                List<Mutation> mutList = null;
                if (!innerMutationMap.ContainsKey(columnFamily))
                {
                    if (SizeHint == null)
                    {
                        mutList = new List<Mutation>();
                    }
                    else
                    {
                        mutList = new List<Mutation>(SizeHint.NumOfColumns);
                    }
                    innerMutationMap.Add(columnFamily, mutList);
                }
                else
                {
                    mutList = innerMutationMap[columnFamily];
                }
                mutList.Add(mutation);
            }
        }
        private Dictionary<string, List<Mutation>> GetInnerMutationMap(K key)
        {
            ByteBuffer keybytes = KeySerializer.ToByteBuffer(key);
            Dictionary<string, List<Mutation>> innerMutationMap = null;
            if (!MutationDictionary.ContainsKey(keybytes))
            {
                innerMutationMap = new Dictionary<string, List<Mutation>>();
                MutationDictionary.Add(keybytes, innerMutationMap);
            }
            else
            {
                innerMutationMap = MutationDictionary[keybytes];
            }
            return innerMutationMap;
        }
        public Dictionary<byte[], Dictionary<string, List<Mutation>>> ConvertMutationMap()
        {
            Dictionary<byte[], Dictionary<string, List<Mutation>>> mutations = new Dictionary<byte[], Dictionary<string, List<Mutation>>>();
            foreach (ByteBuffer bbuff in MutationDictionary.Keys)
            {
                Dictionary<string, List<Mutation>> value = MutationDictionary[bbuff];
                mutations.Add(bbuff.GetBuffer(),value);
            }
            return mutations;
        }
        public BatchMutation<K> makeCopy()
        {
            return new BatchMutation<K>(KeySerializer, MutationDictionary, SizeHint);
        }
        public bool IsEmpty()
        {
            return MutationDictionary.Count == 0;
        }
        public int Count
        {
            get { return MutationDictionary.Count; }
        }
    }
}